package com.atguigu.gmall.realtime.util;

import com.atguigu.gmall.realtime.bean.TransientSink;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @author:bryant
 * @create:2022-05-29
 * @Des: 操作 clickhouse 工具类
 */
public class MyClickHouseUtil {

    //获取clickhouse的sinkfunction的方法
    public static <T> SinkFunction<T> getJdbcSink(String sql) {
        return JdbcSink.sink(
                sql,

                new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement ps, T obj) throws SQLException {
                        //给问号占位符赋值,不知道对象的类型是什么，通过反射获取
                        //反射获取当前类中的属性
                        Field[] filedsArr = obj.getClass().getDeclaredFields();

                        int skipNum = 0;
                        //遍历属性
                        for (int i = 0; i < filedsArr.length; i++) {
                            //获取一个属性对象，包括属性名，属性修饰符等
                            Field field = filedsArr[i];
                            //判断是否被注解修饰
                            TransientSink transientSink = field.getAnnotation(TransientSink.class);
                            if (transientSink != null){
                                skipNum ++;
                                continue;
                            }

                            //设置私有属性访问权限
                            field.setAccessible(true);

                            try {
                                //获取属性的值
                                Object filedValue = field.get(obj);
                                //给占位符赋值
                                ps.setObject(i + 1 - skipNum,filedValue);
                            } catch (IllegalAccessException e) {
                                e.printStackTrace();
                            }

                        }
                    }
                    
                },

                JdbcExecutionOptions.builder()
                        .withBatchSize(5)
                        .build(),

                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(GmallConfig.CLICKHOUSR_DRIVER)
                        .withUrl(GmallConfig.CLICKHOUSR_URL)
                        .build()
        );
    }

}
